package com.naiterui.ehp.bs.agent.websocket;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Service;

import java.util.Optional;

/**
 * websocket集群节点消息处理类
 * redis - list 简易有序队列集群共享
 *
 * @author gaoliangliang
 * @since 2022/2/22
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class WebsocketConsumer {

    private final WebsocketSession websocketSession;

    @Bean
    public MessageListenerAdapter messageListenerAdapter() {
        return new MessageListenerAdapter((MessageListener) (message, pattern) -> {
            log.info("服务集群消息推送处理 message {}", message);
            WebsocketMessage taskMessage = Optional.of(message.getBody()).map(String::new)
                    .map(messageStr -> JSONUtil.toBean(messageStr, WebsocketMessage.class)).orElse(null);
            if (taskMessage == null || StrUtil.hasBlank(taskMessage.getMessage())) {
                return;
            }
            if (WebsocketConstant.ALL_USER.equals(taskMessage.getToUserId())) {
                websocketSession.sendMessage(taskMessage.getMessage());
            } else {
                websocketSession.queueMessage(taskMessage.getToUserId());
            }
        });
    }

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(messageListenerAdapter, new PatternTopic(WebsocketConstant.AGENT_MESSAGE_QUEUE));
        return container;
    }

}
